package com.shujia.spark.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo4UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

    conf.setAppName("stateByKeu")
    conf.setMaster("local[2]")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
      * cache :将rdd的数据保存到内存或者磁盘上
      * checkpoint:: 将rdd的数据持久化到hdfs中
      *
      * 有状态算子需要使用checkpoint将计算的状态持久化到hdfs中
      */
    ssc.checkpoint("data/checkpoint")

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val wordsDS: DStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))


    //reduceByKey： 只统计当前batch的数据,不能进行累加统计
    //val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_+_)

    /**
      * 更新之前的计算状态
      *
      * @param seq    : 当前batch一个key所有的value
      * @param option : 之前的计算状态
      * @return : 返回新的状态
      */
    def updateFun(seq: Seq[Int], option: Option[Int]): Option[Int] = {
      //1、统计当前batch一个单词的数量
      val currCount: Int = seq.sum
      //2、获取前面这个单词的数量
      val lessCount: Int = option.getOrElse(0)
      //3、计算新的单词的数量
      val count: Int = currCount + lessCount

      //4、返回单词的数量
      Option(count)
    }

    /**
      * updateStateByKey: 有状态算子
      * 每一次计算基于前面的计算结果进行累加
      *
      */
    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updateFun)


    countDS.print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
